import { add, isBefore } from "date-fns";
import { WebWorkerBase } from "../../../ts/webworker";
import { MQWorkerMessageType } from "./common";
import { getLogger } from "../../../ts/log";
import { Instance } from "../../../ts";
import * as lodash from "lodash";

interface WorkerMQWebSocketConfigInternal {
  // websocket 服务路径
  url: string;
}

interface MQWebSocketPack {
  topic: string;
  packs: number;
  msgs: { index: number; msg: string }[];
  time: Date;
}

function doUnPack(data: any) {
  const logger = getLogger("ys-ts.mqUnpack");

  const packBuf = Instance.Instance().getOrSetFunc<
    Map<string, MQWebSocketPack>
  >("mq.unpack", () => {
    return new Map<string, MQWebSocketPack>();
  });

  const re = /(.*)\[([0-9]+)-([0-9]+)]$/;
  if (!re.test(data.topic)) {
    return data;
  }

  const match = re.exec(data.topic)!;
  const topic = match[1];
  const index = match[2];
  const packs = match[3];

  // 10秒内没有来包就删除
  if (packBuf.has(topic)) {
    if (isBefore(packBuf.get(topic)!.time, Date.now())) {
      logger(`超时 ${topic}`);
      packBuf.delete(topic);
    }
  }

  if (!packBuf.has(topic)) {
    packBuf.set(topic, {
      topic,
      packs: Number(packs),
      msgs: [],
      time: add(Date.now(), {
        seconds: 10,
      }),
    });
  }

  logger(`正在处理 ${topic}-${index}`);

  const pack = packBuf.get(topic)!;
  pack.msgs.push({
    index: Number(index),
    msg: data.message,
  });

  if (pack.msgs.length >= pack.packs) {
    logger(`正在拼接数据 ${topic}`);
    const n = lodash.sortBy(pack.msgs, ["index"]);
    let message = "";
    n.forEach((value) => {
      message += value.msg;
    });
    packBuf.delete(topic);
    return {
      topic: topic,
      message: message,
    };
  }

  return null;
}

class MQWorker extends WebWorkerBase {
  // websocket 连接对象
  private ws: WebSocket | undefined;
  // 配置参数
  private timer: NodeJS.Timeout | undefined;
  private topics = "";
  private id = 0;

  constructor() {
    super();
    console.log("构造ws对象");
  }

  init() {
    super.init();

    this.addHandler(MQWorkerMessageType.OPEN, (type, msg) => {
      this.logger("启动连接");
      this.doConnect();
    });
  }

  private doConnect() {
    this.ws = new WebSocket(this.cfg.url);

    // 注意这里使用 lambda 表达式让函数内的 this 指向正确的对象
    this.ws.onclose = () => {
      this.onWSClose();
    };
    this.ws.onopen = () => {
      this.onWSOpen();
    };
    this.ws.onmessage = (result: MessageEvent) => {
      this.onWSMessage(result);
    };
  }

  private onWSClose() {
    this.ws = undefined;
    this.postMessage({ type: MQWorkerMessageType.CLOSE, data: {} });
    this.logger(`连接已关闭`);
    this.timer = setTimeout(() => {
      this.doConnect();
    }, 3000);
  }

  private isConnected() {
    return this.ws != undefined;
  }

  private onWSOpen() {
    this.postMessage({ type: MQWorkerMessageType.OPEN, data: {} });
    console.log(`连接已开启`);
  }

  private onWSMessage(result: MessageEvent) {
    let data = JSON.parse(result.data);
    if (!data.message || !data.topic) {
      this.logger(`WebSocket获取的 data存在问题`, result.data);
    }

    this.logger(`接收话题: `, data.topic);

    // 判断 topic 是否存在分包现象
    data = doUnPack(data);
    if (data == null) {
      this.logger(`分包数据没有处理`);
      return;
    }

    const transType = lodash.split(data.topic, ".")[0];
    let msg: any = null;
    switch (transType) {
      case "json":
        msg = JSON.parse(data.message);
        break;
      case "ok":
      case "error":
      case "str":
        msg = data.message;
        break;
      default:
        console.error(`不能解析 ${transType} 类数据`);
        return;
    }

    // 对于 ok.sub 主题处理
    if (data.topic === "ok.sub") {
      this.id = JSON.parse(data.message).id;
      this.logger("订阅ID：", this.id);
    }

    // 数据处理完成后通过 post发送结果
    this.postMessage({
      type: MQWorkerMessageType.MESSAGE,
      data: {
        topic: data.topic,
        msg: msg,
      },
    });
  }
}

const worker = new MQWorker();
worker.init();
